package com.androwit.engine.cassandra;

import java.util.Collection;

import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.Serializer;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.MutationResult;
import me.prettyprint.hector.api.mutation.Mutator;

import com.androwit.engine.ItemInfo;
import com.androwit.engine.MappingPersister;
import com.androwit.engine.mapping.MappingField;
import com.androwit.engine.mapping.MappingFieldsDefiner;
import com.androwit.engine.mapping.MultipleColumnField;

/**
 * Specialzed in hector cassandra data insertion
 * 
 * NOTE: uses Mutator<K> in the <O> arguement of {@link MappingPersister}
 * 
 * @param <K>
 *            key type
 * @param <V>
 *            value type
 * @param <I>
 *            input type
 * 
 * @author hamdi
 */
public class HectorPersister<K, V, I> implements
		MappingPersister<K, V, I, ItemInfo<Mutator<K>, I>> {

	protected Cluster cluster;
	protected Serializer<K> keySerializer;
	protected Keyspace keyspace;
	// Configure Cassandra server list for failover
	protected CassandraHostConfigurator cassandraHostConfigurator;
	protected HectorMappingFieldsDefiner<?, K, V> fieldsDefiner;

	public HectorPersister(String clusterName, String host,
			Serializer<K> keySerializer, String keyspaceName,
			HectorMappingFieldsDefiner<?, K, V> fieldsDefiner) {
		// Create a cluster
		this(HFactory.getOrCreateCluster(clusterName, host), keySerializer,
				keyspaceName, fieldsDefiner);
	}

	public HectorPersister(Cluster cluster, Serializer<K> keySerializer,
			String keyspaceName,
			HectorMappingFieldsDefiner<?, K, V> fieldsDefiner) {
		this(cluster, keySerializer, HFactory.createKeyspace(keyspaceName,
				cluster), fieldsDefiner);
	}

	public HectorPersister(Cluster cluster, Serializer<K> keySerializer,
			Keyspace keyspace, HectorMappingFieldsDefiner<?, K, V> fieldsDefiner) {
		this.cluster = cluster;
		this.keySerializer = keySerializer;
		this.keyspace = keyspace;
		this.fieldsDefiner = fieldsDefiner;
	}

	@Override
	public MutationResult flush(ItemInfo<Mutator<K>, I> output) {
		return output.getOutput().execute();
	}

	@Override
	public ItemInfo<Mutator<K>, I> createOutputItem(I input) {
		return new ItemInfo<Mutator<K>, I>(HFactory.createMutator(keyspace,
				keySerializer), input);
	}

	@Override
	public ItemInfo<Mutator<K>, I> writeField(MappingField<?, K, V> field,
			V value, ItemInfo<Mutator<K>, I> output) {
		return writeField((HectorMappingField<?, K, V>) field, value, output);
	}

	@SuppressWarnings("rawtypes")
	public ItemInfo<Mutator<K>, I> writeField(
			HectorMappingField<?, K, V> field, V value,
			ItemInfo<Mutator<K>, I> item) {

		if (field instanceof MultipleColumnField)
			return writeMultiValue((MultipleColumnField) field, value, item);

		if (value != null) {
			HColumn<K, V> column = HFactory.createColumn(field.getKey(), value,
					keySerializer, field.getValueSerializer());
			item.setOutput(item.getOutput().addInsertion(
					getKey(item.getInput(), fieldsDefiner), getColumnFamily(),
					column));
		}

		return item;
	}

	/**
	 * 
	 * writes a MultiValued item through
	 * {@link HectorMappingField#writeValue(ItemInfo, Object, Object, Serializer)}
	 * override for a different behavior
	 * 
	 * @param field
	 *            field definition
	 * @param value
	 *            he value to write
	 * @param item
	 *            the current line
	 * @return the current line
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	protected ItemInfo<Mutator<K>, I> writeMultiValue(
			MultipleColumnField field, V value, ItemInfo<Mutator<K>, I> item) {
		return (ItemInfo<Mutator<K>, I>) field.writeValue(item,
				getKey(item.getInput(), fieldsDefiner), (Collection) value,
				getColumnFamily());
	}

	@Override
	public Object close(ItemInfo<Mutator<K>, I> item) {
		return null;
	}

	public String getColumnFamily() {
		return fieldsDefiner.getColumnFamily();
	}

	@Override
	public K getKey(I input, MappingFieldsDefiner<?, K, V> definer) {
		return ((HectorMappingFieldsDefiner<?, K, V>) definer).getKey(input);
	};

}
